-- **************************************************************************
--    Project Name:   dwd层-网点中心综合账单 清洗
--    Job Name:       spmi_dwd.dwd_spmn_apack_bill_dt（分区字段：账单产生时间）
--    Description :   从ods层spmi_union_bill表取每天T-1增量更新的数据，T-1 更新的数据划分为三部分，A： 最近30天账单产生的数据， B：最近31-60天账单产生的数据，C：60天之前账单产生的数据，
--                    A部分数据和 spmi_dwd.dwd_spmn_apack_bill_dt 最近30天基于运单号去重，然后回刷回0-30天分区；
--                    B部分数据和  spmi_dwd.dwd_spmn_apack_bill_dt 最近30-60天数据基于运单号去重，然后回刷回31-60天分区
--                    C部分数据直接丢弃，spmi_dwd.dwd_spmn_apack_bill_dt 60天之前的 数据不予更新
--    Author :        songjun
--    date：          2022/5/23
-- **************************************************************************
--    modify by songjun 2022/05/23 上线
-- **************************************************************************

SET hive.merge.mapfiles = true; --  默认true，在map-only任务结束时合并小文件
SET hive.merge.mapredfiles = true; -- 默认false，在map-reduce任务结束时合并小文件
SET hive.merge.size.per.task = 25610001000; -- 默认256M
SET hive.merge.smallfiles.avgsize = 16777216; -- 当输出文件的平均大小小于16m该值时，启动一个独立的map-reduce任务进行文件merge


with tmp_spmi_union_bill_incr as (
select
id
,waybill_no
,settle_network_name
,settle_network_code
,settle_network_id
,financial_center_name
,financial_center_code
,financial_center_id
,waybill_num
,waybill_piece
,cost_weight
,settle_weight
,transfer_cost
,operation_cost
,piece_fee_cost
,distribu_depart_cost
,distribu_opera_cost
,is_package_cost
,product_type_id
,product_type_name
,goods_type_id
,goods_type_name
,transport_type_id
,transport_type_name
,customer_code
,customer_name
,begin_id
,begin_name
,sender_province_id
,sender_province_name
,destination_id
,destination_name
,receiver_province_id
,receiver_province_name
,waybill_time
,sign_time
,send_financial_center_id
,send_financial_center_name
,send_financial_center_code
,send_network_id
,send_network_code
,send_network_name
,sign_financial_center_id
,sign_financial_center_code
,sign_financial_center_name
,sign_network_id
,sign_network_code
,sign_network_name
,operation_network_code
,operation_network_name
,operation_financial_center_code
,operation_financial_center_name
,is_sign
,is_bad_piece
,is_refund
,remarks
,version
,bill_generation_date
,create_time
,update_time
,last_update_time_sync
,trace_id
,customer_fee
,distribu_version
,waybill_weight
,package_charge_weight_source
,center_packge_cost
,distribu_packge_cost
,is_bubble
,retrieve_status
,settle_franchisee_name
,settle_franchisee_code
,settle_franchisee_id
,is_equals_send_piece
,cast(to_date(bill_generation_date) as string) as dt
from spmi_ods.spmi_union_bill where dt  = '{{ execution_date | cst_ds }}'
),
tmp_spmi_union_bill_incr_A as (
select * from tmp_spmi_union_bill_incr where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-30) and '{{ execution_date | cst_ds }}'
),
tmp_spmi_union_bill_incr_B as (
select * from tmp_spmi_union_bill_incr where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-60) and date_add('{{ execution_date | cst_ds }}',-31)
),
tmp_dwd_spmi_union_bill_incr_dt as (
select
id
,waybill_no
,settle_network_name
,settle_network_code
,settle_network_id
,financial_center_name
,financial_center_code
,financial_center_id
,waybill_num
,waybill_piece
,cost_weight
,settle_weight
,transfer_cost
,operation_cost
,piece_fee_cost
,distribu_depart_cost
,distribu_opera_cost
,is_package_cost
,product_type_id
,product_type_name
,goods_type_id
,goods_type_name
,transport_type_id
,transport_type_name
,customer_code
,customer_name
,begin_id
,begin_name
,sender_province_id
,sender_province_name
,destination_id
,destination_name
,receiver_province_id
,receiver_province_name
,waybill_time
,sign_time
,send_financial_center_id
,send_financial_center_name
,send_financial_center_code
,send_network_id
,send_network_code
,send_network_name
,sign_financial_center_id
,sign_financial_center_code
,sign_financial_center_name
,sign_network_id
,sign_network_code
,sign_network_name
,operation_network_code
,operation_network_name
,operation_financial_center_code
,operation_financial_center_name
,is_sign
,is_bad_piece
,is_refund
,remarks
,version
,bill_generation_date
,create_time
,update_time
,last_update_time_sync
,trace_id
,customer_fee
,distribu_version
,waybill_weight
,package_charge_weight_source
,center_packge_cost
,distribu_packge_cost
,is_bubble
,retrieve_status
,settle_franchisee_name
,settle_franchisee_code
,settle_franchisee_id
,is_equals_send_piece
,dt
from spmi_dwd.dwd_spmi_union_bill_dt where dt between date_add('{{ execution_date | cst_ds }}',-60) and  '{{ execution_date | cst_ds }}'
)

insert overwrite table spmi_dwd.dwd_spmi_union_bill_dt partition(dt)
select
 id
 ,waybill_no
 ,settle_network_name
 ,settle_network_code
 ,settle_network_id
 ,financial_center_name
 ,financial_center_code
 ,financial_center_id
 ,waybill_num
 ,waybill_piece
 ,cost_weight
 ,settle_weight
 ,transfer_cost
 ,operation_cost
 ,piece_fee_cost
 ,distribu_depart_cost
 ,distribu_opera_cost
 ,is_package_cost
 ,product_type_id
 ,product_type_name
 ,goods_type_id
 ,goods_type_name
 ,transport_type_id
 ,transport_type_name
 ,customer_code
 ,customer_name
 ,begin_id
 ,begin_name
 ,sender_province_id
 ,sender_province_name
 ,destination_id
 ,destination_name
 ,receiver_province_id
 ,receiver_province_name
 ,waybill_time
 ,sign_time
 ,send_financial_center_id
 ,send_financial_center_name
 ,send_financial_center_code
 ,send_network_id
 ,send_network_code
 ,send_network_name
 ,sign_financial_center_id
 ,sign_financial_center_code
 ,sign_financial_center_name
 ,sign_network_id
 ,sign_network_code
 ,sign_network_name
 ,operation_network_code
 ,operation_network_name
 ,operation_financial_center_code
 ,operation_financial_center_name
 ,is_sign
 ,is_bad_piece
 ,is_refund
 ,remarks
 ,version
 ,bill_generation_date
 ,create_time
 ,update_time
 ,last_update_time_sync
 ,trace_id
 ,customer_fee
 ,distribu_version
 ,waybill_weight
 ,package_charge_weight_source
 ,center_packge_cost
 ,distribu_packge_cost
 ,is_bubble
 ,retrieve_status
 ,settle_franchisee_name
 ,settle_franchisee_code
 ,settle_franchisee_id
 ,is_equals_send_piece
 ,dt
 from (
select * from
(
select * , row_number() over(partition by waybill_no order by  update_time desc) as rnk  from
(select
* from tmp_dwd_spmi_union_bill_incr_dt where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-30) and '{{ execution_date | cst_ds }}'
union all
select
* from tmp_spmi_union_bill_incr_A
) tmp_union  ) tmp_union_rnk where  rnk  = 1

union all
select * from
(
select * , row_number() over(partition by waybill_no order by  update_time desc) as rnk  from
(select
* from tmp_dwd_spmi_union_bill_incr_dt where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-60) and date_add('{{ execution_date | cst_ds }}',-31)
union all
select
* from tmp_spmi_union_bill_incr_B
) tmp_union  ) tmp_union_rnk where  rnk  = 1

) tmp_union_all ;